package com.example.rxjava;

import androidx.annotation.RequiresApi;
import androidx.appcompat.app.AppCompatActivity;

import android.os.Build;
import android.os.Bundle;
import android.util.Log;
import android.util.TimeUtils;
import android.view.View;
import android.widget.TextView;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;

public class MainActivity extends AppCompatActivity {

    private static final String TAG = MainActivity.class.getSimpleName();
    private TextView txt_des;
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        txt_des=findViewById(R.id.txt_des);
    }

    /**
     *  可用Observable.just()  快速创建 & 发送事件更加快捷的实现
     */
    public void observer(View view) {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("1");
                e.onNext("Rx");
                e.onNext("java");
                e.onNext("4");
                e.onNext("5");
//                e.onError(new Throwable("error is access"));
              e.onComplete();
            }
        }).filter(new Predicate<String>() {
            @Override
            public boolean test(String s) throws Exception {
                return "java".equals(s);
            }
        })
                .subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "对onSubscribe事件作出响应");
            }

            @Override
            public void onNext(String s) {
                Log.d(TAG, "对onNext事件" +s+ "作出响应");
                txt_des.setText(s);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "对onError事件作出响应" + e.getMessage());
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "对Complete事件作出响应");
            }
        });
    }

    //无限轮训
    @RequiresApi(api = Build.VERSION_CODES.N)
    public void interval(View view) {
        Observable.interval(2, TimeUnit.SECONDS).doOnNext(new Consumer<Long>() {
            @Override
            public void accept(Long integer) throws Exception {
                Log.d(TAG, "第 " + integer + " 次轮询" );
            }
        }).subscribeOn(Schedulers.io())               // 切换到IO线程进行网络请求
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(TAG, "对onSubscribe事件作出响应");
                    }

                    @Override
                    public void onNext(Long aLong) {
                        Log.e(TAG, "对onNext事件" + aLong + "作出响应"  );
                        String des = "无限轮训-对onNext事件[" + aLong + "]作出响应";
                        txt_des.append(des);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, "对onError事件作出响应" + e.getMessage());
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "对Complete事件作出响应");
                    }
                });
    }


    //有限轮训
   public void intervalsss(View view){
       //表示从0(包括)开始，发射10条数据，发射前延迟2s,然后每隔1s发射一条数据，使用的是Schedulers.trampoline()调度器
            Observable.intervalRange(0,10,2,1,TimeUnit.SECONDS,Schedulers.trampoline())
                    .doOnNext(new Consumer<Long>() {
                        @Override
                        public void accept(Long aLong) throws Exception {
                            Log.d(TAG, "第 " + aLong + " 次轮询" );
                        }
                    }).subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Observer<Long>() {
                        @Override
                        public void onSubscribe(Disposable d) {
                            Log.d(TAG, "对onSubscribe事件作出响应");
                        }

                        @Override
                        public void onNext(Long aLong) {
                            Log.e(TAG, "对onNext事件" + aLong + "作出响应"  );
                            String des = "有限轮训-对onNext事件[" + aLong + "]作出响应";
                            txt_des.append(des);
                        }

                        @Override
                        public void onError(Throwable e) {
                            Log.d(TAG, "对onError事件作出响应" + e.getMessage());
                        }

                        @Override
                        public void onComplete() {
                            Log.d(TAG, "对Complete事件作出响应");
                        }
                    });
   }
}
